package com.example.ranking.listener;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.example.ranking.config.MqExchangeQueueConfig;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * Canal客户端：监听pet_info表的lend_status变更，发送到MQ
 */
@Component
@Slf4j
@RequiredArgsConstructor
public class PetInfoCanalClient implements InitializingBean {
    @Value("${canal.server.host:127.0.0.1}")
    private String canalHost;
    @Value("${canal.server.port:11111}")
    private Integer canalPort;
    @Value("${canal.instance.destination:example}")
    private String destination;
    @Value("${canal.instance.username:}")
    private String username;
    @Value("${canal.instance.password:}")
    private String password;

    private final RabbitTemplate rabbitTemplate;
    private final MqExchangeQueueConfig rabbitMqConfig; // MQ队列配置类

    // 关注的状态：3-运输中，4-在代养人手里
    private static final List<Integer> TARGET_STATUSES = List.of(3, 4);

    @Override
    public void afterPropertiesSet() throws Exception {
        // 1. 建立Canal连接
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(canalHost, canalPort),
                destination, username, password);
        connector.connect();
        // 2. 订阅pet_info表（全量订阅）
        connector.subscribe("pet_service.pet_info");
        // 3. 启动监听线程（异步处理）
        new Thread(() -> {
            while (true) {
                try {
                    // 4. 获取Binlog消息（每次拉取100条）
                    Message message = connector.getWithoutAck(100);
                    long batchId = message.getId();
                    if (batchId == -1 || message.getEntries().isEmpty()) {
                        Thread.sleep(1000); // 无消息时休眠1秒，避免空轮询
                        continue;
                    }
                    // 5. 解析消息，筛选目标状态变更
                    parseCanalMessage(message.getEntries());
                    // 6. 确认消息消费成功
                    connector.ack(batchId);
                } catch (Exception e) {
                    log.error("Canal监听pet_info表异常", e);
                    try {
                        Thread.sleep(5000); // 异常时休眠5秒，避免频繁报错
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                    }
                }
            }
        }, "canal-pet-info-listener").start();
        log.info("Canal客户端启动成功，开始监听pet_info表状态变更");
    }

    /**
     * 解析Canal消息，筛选“状态3/4变更”的记录，发送到MQ
     */
    private void parseCanalMessage(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            // 过滤非行级变更（如DDL语句）
            if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                continue;
            }
            try {
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    // 解析“变更前”和“变更后”的字段
                    String petId = getFieldValue(rowData, "id");
                    String petBreed = getFieldValue(rowData, "pet_breed");
                    Integer oldStatus = getFieldValueAsInt(rowData.getBeforeColumnsList(), "lend_status");
                    Integer newStatus = getFieldValueAsInt(rowData.getAfterColumnsList(), "lend_status");

                    // 筛选条件：状态从“非3/4”变为“3/4”（需加1），或从“3/4”变为“非3/4”（需减1）
                    boolean needUpdate = (isTargetStatus(newStatus) && !isTargetStatus(oldStatus))
                            || (!isTargetStatus(newStatus) && isTargetStatus(oldStatus));
                    if (needUpdate) {
                        // 构建消息体（含宠物ID、品种、变更类型：ADD-加1，REMOVE-减1）
                        PetStatusChangeMsg msg = new PetStatusChangeMsg();
                        msg.setPetId(petId);
                        msg.setPetBreed(petBreed);
                        msg.setChangeType(isTargetStatus(newStatus) ? "ADD" : "REMOVE");
                        // 发送到MQ队列
                        rabbitTemplate.convertAndSend(
                                rabbitMqConfig.getPetStatusExchange(),
                                rabbitMqConfig.getPetStatusRoutingKey(),
                                msg);
                        log.info("发送宠物状态变更消息：{}", msg);
                    }
                }
            } catch (Exception e) {
                log.error("解析Canal消息异常，entry={}", entry, e);
            }
        }
    }

    /**
     * 判断状态是否为目标状态（3/4）
     */
    private boolean isTargetStatus(Integer status) {
        return status != null && TARGET_STATUSES.contains(status);
    }

    /**
     * 获取字段值（优先取afterColumns，无则取beforeColumns）
     */
    private String getFieldValue(CanalEntry.RowData rowData, String fieldName) {
        Integer value = getFieldValueAsInt(rowData.getAfterColumnsList(), fieldName);
        if (value != null) {
            return value.toString();
        }
        value = getFieldValueAsInt(rowData.getBeforeColumnsList(), fieldName);
        return value != null ? value.toString() : null;
    }

    /**
     * 从列列表中获取字段的Integer值
     */
    private Integer getFieldValueAsInt(List<CanalEntry.Column> columns, String fieldName) {
        for (CanalEntry.Column column : columns) {
            if (column.getName().equals(fieldName) && !column.getValue().isEmpty()) {
                return Integer.parseInt(column.getValue());
            }
        }
        return null;
    }

    /**
     * 宠物状态变更消息体
     */
    @lombok.Data
    public static class PetStatusChangeMsg {
        private String petId;
        private String petBreed;
        private String changeType; // ADD-加1，REMOVE-减1
    }
}
